package com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.utils

import org.apache.spark.sql.{DataFrame, SparkSession}

class BrandNewAddToOld {

  /**
    *
    * @param brandAddNewData 新增品牌数据
    * @param brandTableOld 老品牌表
    * @param spark
    * @param brandTableResult 品牌结果表路径
    * @param BrandResultPath 原始数据关联完品牌的结果路径
    * @param sourceData 原始数据
    * @param platformName 平台名称
    * @param year 年
    * @param month 月
    * @param brandNewAddTable 新增品牌表路径
    * @param timeStamp 时间戳
    */
  def newToOld(brandAddNewData:DataFrame,brandTableOld:DataFrame,spark:SparkSession,brandTableResult:String,
               BrandResultPath:String,sourceData:DataFrame,platformName:String,
               year:Int,month:Int,brandNewAddTable:String,timeStamp:String): Unit ={

    brandAddNewData.registerTempTable("brandNewAddTable")
    brandTableOld.registerTempTable("brandOldTable")
//      新增品牌ID如果在历史品牌表中出现过，则把历史品牌名称，品牌来源国，内资外资，老字号标签给新增品牌
      val brandInOld = spark.sql(
        s"""
          |select a.*,
          |'${platformName}'platform,
          |b.brandName_cn as cn,
          |b.brandName_en as en,
          |b.brand_state as state,
          |b.brand_type as type,
          |b.brand_isLaoZiHao as laozihao
          |from
          |brandNewAddTable a
          |join
          |brandOldTable b
          |on a.brandValueId = b.brandValueId
        """.stripMargin)
          .drop("brandName_cn","brandName_en","brand_state","brand_type","brand_isLaoZiHao")
          .withColumnRenamed("cn","brandName_cn")
          .withColumnRenamed("en","brandName_en")
          .withColumnRenamed("state","brand_state")
          .withColumnRenamed("type","brand_type")
          .withColumnRenamed("laozihao","brand_isLaoZiHao")


    // 新增品牌ID如果没有在历史品牌表中出现过，则继续执行以下流程
    val brandNotInOld = spark.sql(
      """
        |select a.*
        |from
        |brandNewAddTable a
        |left join
        |brandOldTable b
        |on a.brandValueId = b.brandValueId
        |where b.brandValueId is null
      """.stripMargin)

    val brandInOldCount = brandInOld.count()
    val brandNotInOldCount = brandNotInOld.count()

//    不全部关联到的，则把关联不到的继续执行以下流程
    if(brandInOldCount > 0 && brandNotInOldCount > 0){
      println("----------------不全部关联到的---------------------------")
      val brand_modify = new brand_modify_util
      val brand_stap_2 = brand_modify.brand_modify_1(brandNotInOld,spark,platformName,year,month,timeStamp)
//      通过case when对正则修改不掉的品牌名称进行修改
      val brandcase = new brandCase
      val brandcaseResult = brandcase.brandCaseWhen(brand_stap_2,spark)
//      将与历史表关联上的，关联不上的以及历史表进行合并
      val result = brandcaseResult.select("brandCcId","brandName","brandName_cn","brandName_en","brandValueId","brand_isLaoZiHao","brand_type","firstCategoryId","platform","timeStamp","brand_state")
        .union(
      brandInOld.select("brandCcId","brandName","brandName_cn","brandName_en","brandValueId","brand_isLaoZiHao","brand_type","firstCategoryId","platform","timeStamp","brand_state"))
        .union(
      brandTableOld.select("brandCcId","brandName","brandName_cn","brandName_en","brandValueId","brand_isLaoZiHao","brand_type","firstCategoryId","platform","timeStamp","brand_state"))
        .dropDuplicates("brandCcId")
      //      输出品牌结果表
      result.repartition(1).write.mode("overwrite").json(brandTableResult)
      val brand = new BrandJoin
      val brandResultData =  brand.brandResult(sourceData,spark,result)
      brandResultData.coalesce(30).write.orc(BrandResultPath)

//      如果全部都能关联到
    }else if(brandNotInOldCount == 0){
      println("----------------全部都能关联到---------------------------")
      newInOld(spark,brandTableResult,sourceData,BrandResultPath,brandInOld,brandTableOld)

//      如果全部都关联不到
    }else if(brandInOldCount == 0){
      println("----------------全部都关联不到---------------------------")
      newNotInOld_def(brandAddNewData,spark,platformName,year,month,
        brandTableOld,brandTableResult,sourceData,BrandResultPath,timeStamp)

    }else{
      println("----------------其他情况---------------------------")
      newNotInOld_def(brandAddNewData,spark,platformName,year,month,
        brandTableOld,brandTableResult,sourceData, BrandResultPath,timeStamp)
    }



  }


  /**
   *
   * @param brandAddNewData 新增品牌数据
   * @param brandTableOld 老品牌表
   * @param spark
   * @param brandTableResult 品牌结果表路径
   * @param BrandResultPath 原始数据关联完品牌的结果路径
   * @param sourceData 原始数据
   * @param platformName 平台名称
   * @param year 年
   * @param month 月
   * @param brandNewAddTable 新增品牌表路径
   * @param timeStamp 时间戳
   */
  def newToOld_newplat(brandAddNewData:DataFrame,spark:SparkSession,brandTableResult:String,
               BrandResultPath:String,sourceData:DataFrame,platformName:String,
               year:Int,month:Int,brandNewAddTable:String,timeStamp:String): Unit ={

    // 新增品牌ID如果没有在历史品牌表中出现过，则继续执行以下流程

     println("----------------全部都关联不到---------------------------")
    newNotInOld_def_newplat(brandAddNewData,spark,platformName,year,month,brandTableResult,sourceData,BrandResultPath,timeStamp)





  }

  /**
    *
    * @param brandAddNewData 新增品牌数据
    * @param spark
    * @param platformName 平台名称
    * @param year 年
    * @param month 月
    * @param brandTable 老品牌表
    * @param brandTableResult 品牌结果表路径
    * @param sourceData 原始数据
    * @param BrandResultPath 原始数据关联完品牌的结果路径
    * @param timeStamp
    */
  def newNotInOld_def(brandAddNewData:DataFrame, spark:SparkSession, platformName:String, year:Int,month:Int, brandTableOld:DataFrame, brandTableResult:String,
                      sourceData:DataFrame, BrandResultPath:String, timeStamp:String): Unit ={

    val brand_modify = new brand_modify_util
    val brand_stap_2 = brand_modify.brand_modify_1(brandAddNewData,spark,platformName,year,month,timeStamp)
    //      通过case when对正则修改不掉的品牌名称进行修改
    val brandcase = new brandCase
    val brandcaseResult = brandcase.brandCaseWhen(brand_stap_2,spark)
    //      将与历史表关联上的，和新增品牌进行合并
    val result = brandcaseResult.select("brandCcId","brandName","brandName_cn","brandName_en","brandValueId","brand_isLaoZiHao","brand_type","firstCategoryId","platform","timeStamp","brand_state")
      .union(
        brandTableOld.select("brandCcId","brandName","brandName_cn","brandName_en","brandValueId","brand_isLaoZiHao","brand_type","firstCategoryId","platform","timeStamp","brand_state"))

    //      输出品牌结果表
    result.repartition(1).write.mode("overwrite").json(brandTableResult)
    val brand = new BrandJoin
    val brandResultData =  brand.brandResult(sourceData,spark,result)
    brandResultData.coalesce(30).write.orc(BrandResultPath)
  }

  /**
   *
   * @param brandAddNewData 新增品牌数据
   * @param spark
   * @param platformName 平台名称
   * @param year 年
   * @param month 月
   * @param brandTableResult 品牌结果表路径
   * @param sourceData 原始数据
   * @param BrandResultPath 原始数据关联完品牌的结果路径
   * @param timeStamp
   */
  def newNotInOld_def_newplat(brandAddNewData:DataFrame, spark:SparkSession, platformName:String, year:Int,month:Int, brandTableResult:String,
                      sourceData:DataFrame, BrandResultPath:String, timeStamp:String): Unit ={

    val brand_modify = new brand_modify_util
    val brand_stap_2 = brand_modify.brand_modify_1(brandAddNewData,spark,platformName,year,month,timeStamp)
    //      通过case when对正则修改不掉的品牌名称进行修改
    val brandcase = new brandCase
    val brandcaseResult = brandcase.brandCaseWhen(brand_stap_2,spark)
    //      将与历史表关联上的，和新增品牌进行合并
    val result = brandcaseResult.select("brandCcId","brandName","brandName_cn","brandName_en","brandValueId","brand_isLaoZiHao","brand_type","firstCategoryId","platform","timeStamp","brand_state")
    //      输出品牌结果表
    println(platformName+"品牌表路径："+brandTableResult)
    result.repartition(1).write.mode("overwrite").json(brandTableResult)
  }
  /**
    *
    * @param spark
    * @param brandTableResult 品牌结果表路径
    * @param sourceData 原始数据
    * @param BrandResultPath 原始数据关联完品牌的结果路径
    */
  def newInOld(spark:SparkSession,brandTableResult:String,sourceData:DataFrame,BrandResultPath:String,brandInOld:DataFrame,brandTableOld:DataFrame): Unit ={

    val result = brandInOld.select("brandCcId","brandName","brandName_cn","brandName_en","brandValueId","brand_isLaoZiHao","brand_type","firstCategoryId","platform","timeStamp","brand_state")
    .union(
      brandTableOld.select("brandCcId","brandName","brandName_cn","brandName_en","brandValueId","brand_isLaoZiHao","brand_type","firstCategoryId","platform","timeStamp","brand_state"))
      .dropDuplicates("brandCcId")
    //输出品牌结果表
    result.repartition(1).write.mode("overwrite").json(brandTableResult)
    val brand = new BrandJoin
    val brandResultData =  brand.brandResult(sourceData,spark,result)
    brandResultData.coalesce(30).write.orc(BrandResultPath)
  }

}